package notify

import (
	"context"
	"encoding/json"
	"github.com/avast/retry-go"
	"github.com/segmentio/kafka-go"
	"slices"
	"wf-deploy-agent/cmd"
	"wf-deploy-agent/conf"
	"wf-deploy-agent/infra"
)

func Broadcast(topic string, raw interface{}) error {
	w := &kafka.Writer{
		Addr:                   kafka.TCP(conf.CF.Kafka.Brokers...),
		Topic:                  topic,
		Balancer:               &kafka.LeastBytes{},
		AllowAutoTopicCreation: true,
	}

	defer func(w *kafka.Writer) {
		err := w.Close()
		if err != nil {
			infra.Logger.Errorln(err)
		}
	}(w)

	bytes, err := json.Marshal(raw)
	if err != nil {
		infra.Logger.Errorln(err)
		return err
	}

	err = w.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte(""),
			Value: bytes,
		},
	)
	if err != nil {
		infra.Logger.Errorln("failed to write messages:", err)
		return err
	}

	if err = w.Close(); err != nil {
		infra.Logger.Errorln("failed to close writer:", err)
		return err
	}

	return nil
}

/*
true表示资源列表包含当前节点会执行命令
false表示资源列表不包含当前节点不会执行命令
*/
func validate(basic *cmd.Command) bool {
	return slices.Contains(basic.Resources, conf.CF.Agent.Identity)
}

func Subscribe(topic string, group string, handle func(*cmd.Command) error, post ...func(*cmd.Command, error) error) error {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  conf.CF.Kafka.Brokers,
		GroupID:  group,
		Topic:    topic,
		MaxBytes: 10e6,
	})

	go func() {
		ctx := context.Background()
		for {
			if err := recover(); err != nil {
				infra.Logger.Errorln("process topic[%s] occurs err, %v", topic, err)
				continue
			}

			//显式提交消息
			m, err := r.FetchMessage(ctx)
			if err != nil {
				infra.Logger.Errorln(err)
				continue
			}

			key := string(m.Key)
			message := string(m.Value)
			infra.Logger.Debugf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, key, message)

			basic := &cmd.Command{}
			err = json.Unmarshal(m.Value, basic)
			if err != nil {
				infra.Logger.Errorf("process basic message from topic[%s] unmarshal failed, raw message:%s, error:%v", topic, message, err)
				continue
			}

			basic.Raw = message

			//验证当前节点是否匹配资源范围
			if !validate(basic) {
				_ = r.CommitMessages(ctx, m)
				continue
			}

			//添加重试机制，此处极其重要
			//err = retry.Do(
			//	func() error {
			//		return handle(basic)
			//	},
			//	retry.OnRetry(func(n uint, err error) {
			//		infra.Logger.Errorf("process message from topic[%s] occurs error and now retrying, raw message:%s, retry count:%d, error:%v", topic, message, n, err)
			//	}),
			//)

			err = handle(basic)

			if err != nil {
				infra.Logger.Errorf("process message from topic[%s] failed, raw message:%s, error:%v", topic, message, err)
			}

			if post != nil && len(post) > 0 {
				_ = retry.Do(
					func() error {
						return post[0](basic, err)
					},
					retry.OnRetry(func(n uint, err error) {
						infra.Logger.Errorf("process message from topic[%s] success, but post failed and now retrying, raw message:%s, retry count:%d, error:%v", topic, message, n, err)
					}),
				)
			}

			if err = r.CommitMessages(ctx, m); err != nil {
				infra.Logger.Errorf("process message from topic[%s] success, but commit failed, raw message:%s, error:%v", topic, message, err)
				continue
			}
		}
	}()

	//if err := r.Close(); err != nil {
	//	infra.Logger.Fatalln(err)
	//	return err
	//}

	//defer r.Close()

	return nil
}
